package com.zhihaoliang.coroutine.buss.channel.func

import com.zhihaoliang.coroutine.util.log
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.channels.ReceiveChannel
import kotlinx.coroutines.channels.produce
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch

/**
 * @author 支豪亮
 * @date 2022/10/27 17:35
 * @des 扇出
 */
object OutFan {

    @OptIn(ExperimentalCoroutinesApi::class)
    private fun CoroutineScope.produceNumbers() = produce {
        var x = 1 // start from 1
        while (true) {
            send(x++) // produce next
            delay(100) // wait 0.1s
        }
    }

    private fun CoroutineScope.launchProcessor(id: Int, channel: ReceiveChannel<Int>) = launch {
        for (msg in channel) {
            log("Processor #$id received $msg",coroutineContext)
        }
    }

    fun fanOut(mainScope:CoroutineScope) = mainScope.launch {
        val producer = produceNumbers()
        repeat(5) { launchProcessor(it, producer) }
        delay(950)
        producer.cancel() // cancel producer coroutine and thus kill them all
    }
}